package com.atguigu.upp.app;

import com.atguigu.upp.beans.TagInfo;
import com.atguigu.upp.beans.TagValueTypeConstant;
import com.atguigu.upp.beans.TaskInfo;
import com.atguigu.upp.beans.TaskTagRule;
import com.atguigu.upp.service.MysqlService;
import com.atguigu.upp.util.UPPUtil;
import lombok.extern.log4j.Log4j;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;

/**
 * Created by Smexy on 2023/3/24
 */
@Log4j
public class SqlTaskExecuteApp
{
    public static void main(String[] args) throws IOException {

        //接收画像平台传入的参数
        String taskId = args[0];
        String doDate = args[1];

    /*    String taskId = "3";
        String doDate = "2022-06-09";*/

        //查询元数据
        SqlSessionFactory sqlSessionFactory = UPPUtil.createSqlSessionFactory("mysqlconfig.xml");
        MysqlService mysqlService = new MysqlService(sqlSessionFactory.openSession(true));

        TaskInfo taskInfo = mysqlService.getTaskInfo(taskId);
        TagInfo tagInfo = mysqlService.getTagInfo(taskId);
        List<TaskTagRule> rules = mysqlService.queryRules(taskId);

        //使用sparksql
        SparkSession sparkSession = UPPUtil.getSparkSession("SqlTaskExecuteApp");
        //准备建表语句(保存标签运行的结果)
        String generateCreateSql = generateCreateSql(tagInfo);
        //执行写入语句即可(向结果表中写入标签运行的结果)
        String insertSql = generateInsertSql(tagInfo, taskInfo, rules, doDate);

        sparkSession.sql(generateCreateSql);
        sparkSession.sql(insertSql);

        log.error("ok");


    }

    /*

    判断要计算的 标签 是否有四级标签
        如果没有，直接算。
        如果有，根据 tagValue和四级标签的映射规则，将tagValue转换为对应四级标签的值。
            以性别为例:  M --- 男性

    insert overwrite table upp221109.TAG_POPULATION_ATTRIBUTE_NATURE_GENDER partition (dt='2022-06-09')
    select
        uid,
       case tagValue
    when 'M' then '男性'
    when 'F' then '女性'
    when 'U' then '未知'
    end tagValue
    from (select id                                uid,
             `if`(isnull(gender), 'U', gender) tagValue
    from gmall.dim_user_zip
    where dt = '9999-12-31'
     ) tmp

     */
    private static String generateInsertSql(TagInfo tagInfo, TaskInfo taskInfo, List<TaskTagRule> rules, String doDate) {

        //声明模版
        String template = " insert overwrite table %s.%s partition (dt='%s') " +
            "  select uid, %s from (  %s )tmp  ";

        String db = UPPUtil.getValue("updbname");
        //生成case sql
        String casesql = "";
        //当前标签是一个有四级标签的标签，有映射规则
        if (rules.size() > 0) {
            casesql = " case tagValue  " + rules.stream()
                                                .map(rule -> String.format("  when '%s' then '%s' ", rule.getQueryValue(), rule.getSubTagValue()))
                                                .collect(Collectors.joining(" ")) + "  end tagValue ";
        } else {
            //当前标签没有四级标签
            casesql = " tagValue ";
        }

        //从taskInfo中获取平台上填写的 任务sql。 需要处理！ 去除最后的; 把dodate占位符替换为当天要计算的业务日期
        String sql = taskInfo.getTaskSql().replace(";", "").replace("dodate", doDate);

        //格式化
        String insertSql = String.format(template, db, tagInfo.getTagCode(), doDate, casesql, sql);

        log.warn(insertSql);

        return insertSql;
    }

    /*
  create table if not exists ①.②(
     uid string,
     tagValue ③
 )
 comment '④'
 partitioned by (dt string)
 location '⑤⑥';
     */
    private static String generateCreateSql(TagInfo tagInfo) {

        //java程序中 %s 指字符串格式的占位符
        String template = "  create table if not exists %s.%s(   uid string, tagValue %s )  comment '%s' " +
            "   partitioned by (dt string) location '%s/%s'  ";

        //填充
        String db = UPPUtil.getValue("updbname");

        //根据 标签的 tag_value_type确定
        String tagValueType = "";
        switch (tagInfo.getTagValueType()) {
            case TagValueTypeConstant.TAG_VALUE_TYPE_LONG:
                tagValueType = "bigint";
                break;
            case TagValueTypeConstant.TAG_VALUE_TYPE_DECIMAL:
                tagValueType = "decimal(16,2)";
                break;
            case TagValueTypeConstant.TAG_VALUE_TYPE_STRING:
                tagValueType = "string";
                break;
            case TagValueTypeConstant.TAG_VALUE_TYPE_DATE:
                tagValueType = "string";
                break;
        }

        //获取表存储的父目录
        String hdfsPath = UPPUtil.getValue("hdfsPath");

        //填充模版中的占位符
        String createTableSql = String.format(template, db, tagInfo.getTagCode(), tagValueType, tagInfo.getTagName(), hdfsPath, tagInfo.getTagCode());

        log.warn(createTableSql);

        return createTableSql;
    }
}
